package com.hub.realtime.flinkshims.core;

import com.alibaba.fastjson.JSON;
import com.hub.realtime.common.model.catalog.FlinkCatalog;
import com.streamxhub.streamx.common.util.DeflaterUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogDatabaseImpl;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.module.Module;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.table.types.AbstractDataType;

import java.time.LocalDateTime;
import java.util.Optional;
import java.util.function.Function;

import java.util.*;

/**
 * flink table api
 */
@Slf4j
public abstract class FlinkTableBase implements TableEnvironment {

    protected ParameterTool parameter;
    protected TableEnvironment tableEnv;
    protected StreamExecutionEnvironment senv;

    public FlinkTableBase(ParameterTool parameter, StreamExecutionEnvironment senv, TableEnvironment tableEnv) {
        this.parameter = parameter;
        this.tableEnv = tableEnv;
        this.senv = senv;
    }

    public void start(StatementSet statementSet) {
        String jobName = parameter.get("jobName");
        tableEnv.getConfig().getConfiguration().set(PipelineOptions.NAME, jobName);
        try {
            String logo = "FlinkTable " + jobName + " Starting.....";
            printLogo(logo);
            if (statementSet != null) {
                log.info("====" + logo);
                TableResult tableResult = statementSet.execute();
                log.info("====执行完成" + JSON.toJSONString(tableResult));
                if (tableResult == null || tableResult.getJobClient().get() == null
                        || tableResult.getJobClient().get().getJobID() == null) {
                    throw new RuntimeException("任务运行失败 没有获取到JobID");
                }
                log.info("=======flink sql启动成功，JobId: " + tableResult.getJobClient().get().getJobID().toHexString());
            } else {
                if (senv != null) {
                    log.info("=======开始启动flink任务======");
                    JobExecutionResult execute = senv.execute(jobName);
                    if (execute != null && execute.getJobID() != null) {
                        log.info("=======flink sql启动成功，JobId: " + execute.getJobID().toHexString());
                    }
                }
            }
        } catch (Exception exception) {
            log.error("执行sql出错：" + exception.getMessage());
            exception.printStackTrace();
            throw new IllegalArgumentException(exception);
        }

    }

    public void handelSql() {

        /**
         *  String unzipString = DeflaterUtils.unzipString(config.getContent());
         *             String encode = Base64.getEncoder().encodeToString(unzipString.getBytes());
         *             config.setContent(encode);
         */
        log.info("===========传进来的参数列表：" + JSON.toJSONString(parameter));
        String sqlText = parameter.get("sql");
        sqlText = DeflaterUtils.unzipString(sqlText);
        // tableEnv.createStatementSet()

        /**
         * 创建hive
         */
        String catalog = parameter.get("catalog");
        if (StringUtils.isNotEmpty(catalog)) {
            FlinkCatalog flinkCatalog = JSON.parseObject(catalog, FlinkCatalog.class);
            // 1. 创建HiveCatalog
            HiveCatalog hiveCatalog = new HiveCatalog(flinkCatalog.getCatalogName(), null
                    , flinkCatalog.getHiveConfDir());
            //2.注册HiveCatalog
            Optional<Catalog> catalogOptional = tableEnv.getCatalog(flinkCatalog.getCatalogName());
            if (!catalogOptional.isPresent()) {
                log.info("===catalog不存在注册：" + flinkCatalog.getCatalogName());
                tableEnv.registerCatalog(flinkCatalog.getCatalogName(), hiveCatalog);
            } else {
                log.info("===catalog存在不注册：" + flinkCatalog.getCatalogName());
            }
            Optional<Catalog> catalogAdd = tableEnv.getCatalog(flinkCatalog.getCatalogName());
            if (catalogAdd.isPresent()) {
                Catalog currentCatalog = catalogAdd.get();
                //tableEnv.useCatalog(flinkCatalog.getCatalogName());
                if (!currentCatalog.databaseExists(flinkCatalog.getDatabaseName())) {
                    log.info("===表不存在创建：" + flinkCatalog.getDatabaseName());
                    sqlText= "use catalog ".concat(flinkCatalog.getCatalogName()).concat(";\n")
                            .concat("CREATE DATABASE IF NOT EXISTS ").concat(flinkCatalog.getDatabaseName())
                            .concat(";\n")
                            .concat("use ").concat(flinkCatalog.getDatabaseName()).concat(";\n")
                            .concat(sqlText);
                } else {
                    log.info("===表存在不创建创建：" + flinkCatalog.getDatabaseName());
                    sqlText= "use catalog ".concat(flinkCatalog.getCatalogName()).concat(";\n")
                            .concat("use ").concat(flinkCatalog.getDatabaseName()).concat(";\n")
                            .concat(sqlText);
                }
            }
        }
        log.info("====执行的sql：" + sqlText);
        StatementSet statementSet = sql(sqlText);
        start(statementSet);
    }

    /**
     * 构建方法
     * 执行sql
     * 执行
     */

    /**
     * 处理sql不返回结果
     *
     * @param sql
     */
    public StatementSet sql(String sql) {
        return FlinkSqlExecutor.executeSql(sql, tableEnv, null);
    }


    /**
     * 返回执行结果
     *
     * @param sql
     * @param callBackFunction
     */
    public void sqlWithCallBack(String sql, Function<String, String> callBackFunction) {
        FlinkSqlExecutor.executeSql(sql, tableEnv, callBackFunction);
    }

    /**
     * 开始执行程序
     *
     * @param jobName
     * @return
     */
    @Override
    public JobExecutionResult execute(String jobName) {
        try {
            String logo = "FlinkTable " + jobName + " Starting.....";
            printLogo(logo);
            if (senv != null) {
                log.info("=======开始启动flink任务======");
                return senv.execute(jobName);
            } else {
                return null;
            }

        } catch (Exception ex) {
            log.error("sql执行失败：" + ex.getMessage());
            ex.printStackTrace();
            throw new IllegalArgumentException("sql执行失败：" + ex.getMessage());
        }
    }


    @Override
    public Table fromValues(Object... values) {
        return tableEnv.fromValues(values);
    }

    @Override
    public Table fromValues(AbstractDataType<?> rowType, Object... values) {
        return tableEnv.fromValues(rowType, values);
    }

    @Override
    public Table fromValues(Expression... values) {
        return tableEnv.fromValues(values);
    }

    @Override
    public Table fromValues(AbstractDataType<?> rowType, Expression... values) {
        return tableEnv.fromValues(rowType, values);
    }

    @Override
    public Table fromValues(Iterable<?> values) {
        return tableEnv.fromValues(values);
    }

    @Override
    public Table fromValues(AbstractDataType<?> rowType, Iterable<?> values) {
        return tableEnv.fromValues(rowType, values);
    }

    @Override
    public void registerCatalog(String catalogName, Catalog catalog) {
        tableEnv.registerCatalog(catalogName, catalog);
    }


    @Override
    public Optional<Catalog> getCatalog(String catalogName) {
        return tableEnv.getCatalog(catalogName);
    }


    @Override
    public void loadModule(String moduleName, Module module) {
        tableEnv.loadModule(moduleName, module);
    }

    @Override
    public void unloadModule(String moduleName) {
        tableEnv.unloadModule(moduleName);
    }

    @Override
    public void createTemporarySystemFunction(String name, Class<? extends UserDefinedFunction> functionClass) {
        tableEnv.createTemporarySystemFunction(name, functionClass);
    }

    @Override
    public void createTemporarySystemFunction(String name, UserDefinedFunction functionInstance) {
        tableEnv.createTemporarySystemFunction(name, functionInstance);
    }

    @Override
    public boolean dropTemporarySystemFunction(String name) {
        return tableEnv.dropTemporarySystemFunction(name);
    }

    @Override
    public void createFunction(String path, Class<? extends UserDefinedFunction> functionClass) {
        tableEnv.createFunction(path, functionClass);
    }

    @Override
    public void createFunction(String path, Class<? extends UserDefinedFunction> functionClass, boolean ignoreIfExists) {
        tableEnv.createFunction(path, functionClass, ignoreIfExists);
    }

    @Override
    public boolean dropFunction(String path) {
        return tableEnv.dropFunction(path);
    }

    @Override
    public void createTemporaryFunction(String path, Class<? extends UserDefinedFunction> functionClass) {
        tableEnv.createTemporaryFunction(path, functionClass);
    }

    @Override
    public void createTemporaryFunction(String path, UserDefinedFunction functionInstance) {
        tableEnv.createTemporaryFunction(path, functionInstance);
    }

    @Override
    public boolean dropTemporaryFunction(String path) {
        return tableEnv.dropTemporaryFunction(path);
    }


    @Override
    public void createTemporaryView(String path, Table view) {
        tableEnv.createTemporaryView(path, view);
    }

    @Override
    public Table from(String path) {
        return tableEnv.from(path);
    }

    @Override
    public String[] listCatalogs() {
        return tableEnv.listCatalogs();
    }

    @Override
    public String[] listModules() {
        return tableEnv.listModules();
    }


    @Override
    public String[] listDatabases() {
        return tableEnv.listDatabases();
    }

    @Override
    public String[] listTables() {
        return tableEnv.listTables();
    }


    @Override
    public String[] listViews() {
        return tableEnv.listViews();
    }


    @Override
    public String[] listTemporaryTables() {
        return tableEnv.listTemporaryTables();
    }

    @Override
    public String[] listTemporaryViews() {
        return tableEnv.listTemporaryViews();
    }

    @Override
    public String[] listUserDefinedFunctions() {
        return tableEnv.listUserDefinedFunctions();
    }

    @Override
    public String[] listFunctions() {
        return tableEnv.listFunctions();
    }

    @Override
    public boolean dropTemporaryTable(String path) {
        return tableEnv.dropTemporaryTable(path);
    }


    @Override
    public boolean dropTemporaryView(String path) {
        return tableEnv.dropTemporaryView(path);
    }

    @Override
    public String explainSql(String statement, ExplainDetail... extraDetails) {
        return tableEnv.explainSql(statement, extraDetails);
    }


    @Override
    public Table sqlQuery(String query) {
        return tableEnv.sqlQuery(query);
    }

    @Override
    public TableResult executeSql(String statement) {
        return tableEnv.executeSql(statement);
    }


    @Override
    public String getCurrentCatalog() {
        return tableEnv.getCurrentCatalog();
    }


    @Override
    public void useCatalog(String catalogName) {
        this.unloadModule(catalogName);
    }

    @Override
    public String getCurrentDatabase() {
        return tableEnv.getCurrentDatabase();
    }

    @Override
    public void useDatabase(String databaseName) {
        tableEnv.useDatabase(databaseName);
    }

    @Override
    public TableConfig getConfig() {
        return tableEnv.getConfig();
    }

    @Override
    public StatementSet createStatementSet() {
        return tableEnv.createStatementSet();
    }


    /**
     * 过时的api
     *
     * @param source
     * @return
     */

    @Deprecated
    @Override
    public Table fromTableSource(TableSource<?> source) {
        return tableEnv.fromTableSource(source);
    }

    @Deprecated
    @Override
    public void registerFunction(String name, ScalarFunction function) {
        tableEnv.registerFunction(name, function);
    }


    @Deprecated
    @Override
    public void registerTable(String name, Table table) {
        tableEnv.registerTable(name, table);
    }

    @Deprecated
    @Override
    public Table scan(String... tablePath) {
        return tableEnv.scan(tablePath);
    }


    @Deprecated
    @Override
    public void insertInto(String targetPath, Table table) {
        tableEnv.insertInto(targetPath, table);
    }

    @Deprecated
    @Override
    public void insertInto(Table table, String sinkPath, String... sinkPathContinued) {
        tableEnv.insertInto(table, sinkPath, sinkPathContinued);
    }

    @Deprecated
    @Override
    public String explain(Table table) {
        return tableEnv.explain(table);
    }

    @Deprecated
    @Override
    public String explain(boolean extended) {
        return tableEnv.explain(extended);
    }


    @Deprecated
    @Override
    public String explain(Table table, boolean extended) {
        return tableEnv.explain(table, extended);
    }


    @Deprecated
    @Override
    public String[] getCompletionHints(String statement, int position) {
        return tableEnv.getCompletionHints(statement, position);
    }


    @Deprecated
    @Override
    public void sqlUpdate(String stmt) {
        tableEnv.sqlUpdate(stmt);
    }

    private void printLogo(String info) {
        String printInfo = "          Info      :".concat(info + "                          ");
        System.out.println("                          ▒▓██▓██▒                          ");
        System.out.println("                      ▓████▒▒█▓▒▓███▓▒                      ");
        System.out.println("                   ▓███▓░░        ▒▒▒▓██▒  ▒                ");
        System.out.println("                 ░██▒   ▒▒▓▓█▓▓▒░      ▒████                ");
        System.out.println("                 ██▒         ░▒▓███▒    ▒█▒█▒               ");
        System.out.println("                   ░▓█            ███   ▓░▒██               ");
        System.out.println("                     ▓█       ▒▒▒▒▒▓██▓░▒░▓▓█               ");
        System.out.println("                   █░ █   ▒▒░       ███▓▓█ ▒█▒▒▒            ");
        System.out.println("                   ████░   ▒▓█▓      ██▒▒▒ ▓███▒            ");
        System.out.println("                ░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░            ");
        System.out.println("          ▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒             ");
        System.out.println("         ███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒              ");
        System.out.println("       ░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒            ");
        System.out.println("      ███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░          ");
        System.out.println("     ██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓        ");
        System.out.println("  ▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒      ");
        System.out.println("  ▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒     ");
        System.out.println("  ▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█     ");
        System.out.println("  ██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █     ");
        System.out.println(" ▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓    ");
        System.out.println(" █▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓    ");
        System.out.println(" ██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓     ");
        System.out.println(" ▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒      ");
        System.out.println("  ██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒     ");
        System.out.println("  ▓█   ▒█▓   ░     █░                ▒█              █▓     ");
        System.out.println("   █▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░    ");
        System.out.println("    █▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█    ");
        System.out.println("     ██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓    ");
        System.out.println("      ▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██    ");
        System.out.println("       ░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓    ");
        System.out.println("         ░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒     ");
        System.out.println("             ▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░       ");
        System.out.println("                                                            ");
        System.out.println("                                                            ");
        System.out.println(printInfo);
        System.out.println("                                                            ");
        System.out.println("             Flink        sql执行，时间：" + LocalDateTime.now());
    }
}
